/**
 * 
 */
package io.gitee.falllee.dao.core.table;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.calcite.linq4j.Linq4j;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import io.gitee.falllee.dao.core.HeaderBuilder;
import io.gitee.falllee.dao.model.DaoConst;
import io.gitee.falllee.dao.model.DaoException;
import io.gitee.falllee.dao.persistence.DataMapper;
import io.gitee.falllee.dao.util.CommonUtils;
import io.gitee.falllee.dao.util.FilterPlus;

/**
 * @author batty
 * @date 2022-06-27
 *
 */
public class RestoreBuilder extends HeaderBuilder {

    private TableBuilder tableBuilder;
    
    private DataMapper dataMapper;
    
    private CacheBuilder cacheBuilder;
    
    private MessageBuilder messageBuilder;

    private String tableName;

    private String keyField;
    
    RestoreBuilder(TableBuilder parent) {
        super(parent.getDaoConfig());
        this.tableBuilder = parent;
        this.dataMapper = parent.getDataMapper();
        this.cacheBuilder = parent.cacheBuilder();
        this.messageBuilder = parent.messageBuilder();
        this.tableName = parent.getTableName();
        this.keyField = parent.getKeyField();
    }
    
    public RestoreBuilder header(String key, String value) {
        if (headerParamsMap == null) {
            headerParamsMap = new HashMap<>(16);
        }
        headerParamsMap.put(key, value);
        return this;
    }
    
    public RestoreFilterBuilder filter(FilterPlus filter) {
        return new RestoreFilterBuilder(this, daoConfig, filter);
    }
    
    public Boolean restoreByKey(Object keyValue) {
        if (keyValue == null) {
            throw new DaoException("restoreByKey keyValue can not be empty");
        }
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        map.put(DaoConst.FILTER, getParamExpression(keyField));
        map.put(keyField, keyValue);
        setEnableFlag(map, false);
        setUpdateInfo(map);
        Integer res = dataMapper.restore(map);
        if (res > 0) {
            // 如果启用缓存，则删除缓存数据
            cacheBuilder.clearCache(keyValue);
            // 发送Kafka消息
            messageBuilder.sendMessage(keyValue, DaoConst.RESTORE);
        }
        return res >= 0;
    }

    public Boolean restoreByKeyList(List<? extends Object> keyValueList) {
        List<? extends Object> tempKeyValueList = Linq4j.asEnumerable(keyValueList).where(p -> !ObjectUtils.isEmpty(p))
            .distinct().toList();
        if (CollectionUtils.isEmpty(tempKeyValueList)) {
            throw new DaoException("restoreByKeyList keyValueList can not be empty");
        }
        Map<String, Object> map = new HashMap<>(16);
        map.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        map.put(DaoConst.TABLE_NAME, tableName);
        setEnableFlag(map, false);
        setUpdateInfo(map);
        int max = DaoConst.SELECT_MAX_SIZE.intValue();
        int total = tempKeyValueList.size();
        int num = total / max;
        int left = total % max;
        num = left > 0 ? num + 1 : num;
        Integer res = 0;
        for (int i = 0; i < num; i++) {
            List<? extends Object> subKeyList = CommonUtils.getSubList(tempKeyValueList, max, i);
            String filter = getKeyListFilter(subKeyList, keyField, map);
            map.put(DaoConst.FILTER, filter);
            res += dataMapper.restore(map);
        }
        if (res > 0) {
            // 如果启用缓存，则删除缓存数据
            cacheBuilder.clearCache(tempKeyValueList);
            // 发送Kafka消息
            messageBuilder.sendMessage(tempKeyValueList, DaoConst.RESTORE_LIST);
        }
        return res >= 0;
    }

    Boolean restore(FilterPlus filterPlus) {
        String filter = filterPlus.getParamFilter();
        if (!StringUtils.hasLength(filter)) {
            throw new DaoException("restore filter can not be empty");
        }
        Map<String, Object> map = filterPlus.getParamMap();
        if (map.containsKey(keyField)) {
            throw new DaoException("please use restoreByKey");
        }
        Map<String, Object> countMap = new HashMap<>(16);
        countMap.putAll(map);
        countMap.remove(DaoConst.FIELD);
        countMap.put(DaoConst.DB_TYPE, daoConfig.getDbType());
        countMap.put(DaoConst.TABLE_NAME, tableName);
        countMap.put(DaoConst.FILTER, filter);
        setUpdateInfo(countMap);
        // 由于要查询失效数据数量，这里设置失效条件
        setEnableFlagInvalid(countMap);
        // 先查待还原的数量，分页进行还原
        long count = dataMapper.selectCount(countMap);
        long num = count / DaoConst.SELECT_MAX_SIZE;
        long left = count % DaoConst.SELECT_MAX_SIZE;
        num = left > 0 ? num + 1 : num;
        long rows = DaoConst.SELECT_MAX_SIZE;
        List<Object> keyList = new ArrayList<>();
        for (long i = 0; i < num; i++) {
            long end = i == num - 1 ? count : (i + 1) * rows;
            keyList.addAll(tableBuilder.getKeyList(countMap, keyField, i * rows, rows, end));
        }
        if (CollectionUtils.isEmpty(keyList)) {
            return true;
        }
        return restoreByKeyList(keyList);
    }
}
